const amqp = require('amqplib');
const config = require('../config');

class RabbitMQClient {
    constructor() {
        this.connection = null;
        this.channel = null;
        this.reconnectAttempts = 0;
        this.maxReconnectAttempts = 10;
        this.reconnectInterval = 5000;
        this.queues = new Set(); // 记录已声明的队列
        this.consumers = new Map(); // 记录消费者回调
    }

    // 获取连接字符串
    async getConnectionString() {
        const { protocol, hostname, port, username, password } = config.rabbitMQ;
        return `${protocol}://${username}:${password}@${hostname}:${port}`;
    }
    // 初始化连接
    async initialize() {
        try {
            if (this.connection && this.channel) {
                return this.channel;
            }

            this.connection = await amqp.connect(await this.getConnectionString());
            this.channel = await this.connection.createChannel();

            // 重置重连次数
            this.reconnectAttempts = 0;

            // 连接错误处理
            this.connection.on('error', (error) => {
                console.error('RabbitMQ 连接错误:', error);
                this.handleConnectionFailure();
            });

            // 连接关闭处理
            this.connection.on('close', () => {
                console.log('RabbitMQ 连接关闭');
                this.handleConnectionFailure();
            });

            // 通道错误处理
            this.channel.on('error', (error) => {
                console.error('RabbitMQ 通道错误:', error);
                this.handleConnectionFailure();
            });

            // 通道关闭处理
            this.channel.on('close', () => {
                console.log('RabbitMQ 通道关闭');
                this.handleConnectionFailure();
            });

            console.log('RabbitMQ 连接成功');

            // 重新声明队列和消费者
            await this.restoreQueuesAndConsumers();

            return this.channel;
        } catch (error) {
            console.error('RabbitMQ 初始化失败:', error);
            this.handleConnectionFailure();
            throw error;
        }
    }
    // 连接失败处理
    async handleConnectionFailure() {
        this.connection = null;
        this.channel = null;
        if (this.reconnectAttempts < this.maxReconnectAttempts) {
            this.reconnectAttempts++;
            console.log(`尝试第 ${this.reconnectAttempts} 次重连...`);
            setTimeout(() => this.initialize(), this.reconnectInterval);
        } else {
            console.error('达到最大重连次数，停止重连');
        }
    }
    // 重新声明队列和消费者
    async restoreQueuesAndConsumers() {
        // 重新声明所有队列
        for (const queue of this.queues) {
            await this.channel.assertQueue(queue, { durable: true });
        }

        // 重新启动所有消费者
        for (const [queue, callback] of this.consumers) {
            await this.consume(queue, callback);
        }
    }
    // 获取通道
    async getChannel() {
        if (!this.channel) {
            await this.initialize();
        }
        return this.channel;
    }
    // 发布消息
    async publish(queue, message, options = {}) {
        try {
            const channel = await this.getChannel();

            // 记录队列
            this.queues.add(queue);

            // 确保队列存在
            await channel.assertQueue(queue, { durable: true });

            const messageBuffer = Buffer.from(JSON.stringify(message));
            const publishOptions = {
                persistent: true,
                ...options
            };
            return channel.sendToQueue(queue, messageBuffer, publishOptions);
        } catch (error) {
            console.error('消息发布失败:', error);
            throw error;
        }
    }
    // 消费消息
    async consume(queue, callback, options = {}) {
        try {
            const channel = await this.getChannel();

            // 记录队列和消费者
            this.queues.add(queue);
            this.consumers.set(queue, callback);

            // 确保队列存在
            await channel.assertQueue(queue, { durable: true });

            // 设置预取数量
            await channel.prefetch(options.prefetch || 1);

            return channel.consume(queue, (msg) => {
                if (msg) {
                    try {
                        const content = JSON.parse(msg.content.toString());
                        callback(content);
                        channel.ack(msg);
                    } catch (error) {
                        console.error('消息处理失败:', error);
                        // 根据业务需求决定是否重新入队
                        channel.nack(msg, false, true);
                    }
                }
            }, options);
        } catch (error) {
            console.error('消费者设置失败:', error);
            throw error;
        }
    }
    // 关闭连接
    async shutdown() {
        try {
            if (this.channel) {
                await this.channel.close();
            }
            if (this.connection) {
                await this.connection.close();
            }
            this.channel = null;
            this.connection = null;
            console.log('RabbitMQ 连接已关闭');
        } catch (error) {
            console.error('关闭 RabbitMQ 连接失败:', error);
            throw error;
        }
    }
}

// 创建单例
const rabbitMQ = new RabbitMQClient();

// 退出
process.on('SIGINT', async () => {
    await rabbitMQ.shutdown();
    process.exit(0);
});

process.on('SIGTERM', async () => {
    await rabbitMQ.shutdown();
    process.exit(0);
});

module.exports = rabbitMQ;
